package drds.binlog.parse.driver.socket;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * 实现channel的管理（监听连接、读数据、回收） 2016-12-28
 */
@Slf4j
@SuppressWarnings({"rawtypes"})
public abstract class NettySocketChannels
{


    private static EventLoopGroup eventLoopGroup = new NioEventLoopGroup();                              // 非阻塞IO线程组
    private static Bootstrap bootstrap = new Bootstrap();                                      // 主
    private static Map<Channel, SocketChannel> channelToSocketChannelMap = new ConcurrentHashMap<Channel, SocketChannel>();

    static
    {
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                // 如果是延时敏感型应用，建议关闭Nagle算法
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                //
                .handler(new ChannelInitializer()
                {
                    @Override
                    protected void initChannel(Channel channel) throws Exception
                    {
                        channel.pipeline().addLast(new Handler());// 命令过滤和handler添加管理
                    }
                });
    }

    public static SocketChannel open(SocketAddress socketAddress) throws Exception
    {

        ChannelFuture channelFuture = bootstrap.connect(socketAddress).sync();
        SocketChannel socketChannel = null;
        if (channelFuture.isSuccess())
        {
            channelFuture.channel().pipeline().get(Handler.class).countDownLatch.await();
            socketChannel = channelToSocketChannelMap.get(channelFuture.channel());
        }
        if (null == socketChannel)
        {
            throw new IOException("can't create nettySocketChannel!");
        }

        return socketChannel;
    }

    public static class Handler extends SimpleChannelInboundHandler<ByteBuf>
    {
        private final CountDownLatch countDownLatch = new CountDownLatch(1);
        private NettySocketChannel nettySocketChannel = null;


        @Override
        public void channelActive(ChannelHandlerContext channelHandlerContext) throws Exception
        {
            nettySocketChannel = new NettySocketChannel();
            nettySocketChannel.setChannel(channelHandlerContext.channel());
            channelToSocketChannelMap.put(channelHandlerContext.channel(), nettySocketChannel);
            countDownLatch.countDown();
            super.channelActive(channelHandlerContext);
        }

        @Override
        public void channelInactive(ChannelHandlerContext channelHandlerContext) throws Exception
        {
            nettySocketChannel.setChannel(null);
            channelToSocketChannelMap.remove(channelHandlerContext.channel());// 移除
            super.channelInactive(channelHandlerContext);
        }

        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception
        {
            if (nettySocketChannel != null)
            {
                nettySocketChannel.write(byteBuf);
            } else
            {
                log.error("no nettySocketChannel available.");
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) throws Exception
        {
            log.error("exceptionCaught", throwable);
            channelHandlerContext.close();
        }
    }
}
